package com.bw.gmall.realtime.app.dws.dws_02;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTradeWarningOrderByShop {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 读取DWD层支付成功明细数据（包含shop_id和order_id）
        tableEnv.executeSql("" +
                "CREATE TABLE dwd_trade_pay_detail_suc (\n" +
                "    order_id STRING,\n" +
                "    order_detail_id STRING,\n" +
                "    shop_id STRING,\n" +
                "    callback_time TIMESTAMP(3),  -- 支付时间（转为TIMESTAMP便于计算）\n" +
                "    source_type_code STRING,\n" +
                "    row_op_ts TIMESTAMP_LTZ(3),  -- 事件时间戳\n" +
                "    WATERMARK FOR row_op_ts AS row_op_ts - INTERVAL '5' SECOND  -- 水位线\n" +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_pay_detail_suc", "dws_warning_group"));

        // 3. 读取订单明细变更数据（监控source_type_code更新）
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("topic_db_250703"));
        tableEnv.executeSql("" +
                "CREATE VIEW order_detail_update AS\n" +
                "SELECT\n" +
                "    data['id'] order_detail_id,\n" +  // 订单明细ID，关联支付表
                "    data['order_id'] order_id,\n" +
                "    data['source_type_code'] new_source_type_code,\n" +  // 更新后的code
                "    cast(proc_time as timestamp(3)) update_time  -- 更新时间（转为TIMESTAMP）\n" +
                "FROM topic_db\n" +
                "WHERE `table` = 'order_detail'\n" +  // 订单明细表
                "  AND `type` = 'update'\n" +  // 只处理更新操作
                "  AND data['source_type_code'] IS NOT NULL");  // 确保code字段存在


        // 4. 定义输出表（只保留订单ID和数量，按店铺分组）
        tableEnv.executeSql("" +
                "CREATE TABLE dws_trade_warning_order (\n" +
                "    shop_id STRING,\n" +  // 分组维度：店铺ID
                "    warning_order_count BIGINT,\n" +  // 警告订单数
                "    warning_order_ids ARRAY<STRING>,  // 警告订单ID列表\n" +
                "    window_end TIMESTAMP(3),  // 统计窗口结束时间\n" +
                "    PRIMARY KEY (shop_id, window_end) NOT ENFORCED\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dws_trade_warning_order"));

        // 5. 核心SQL：统计超过3天未将source_type_code改为2402的订单
        tableEnv.executeSql("" +
                "INSERT INTO dws_trade_warning_order\n" +
                "SELECT\n" +
                "    pay.shop_id,\n" +
                "    COUNT(DISTINCT pay.order_id) AS warning_order_count,\n" +  // 去重统计订单数
                "    COLLECT_SET(pay.order_id) AS warning_order_ids,\n" +  // 收集订单ID列表
                "    TUMBLE_END(pay.row_op_ts, INTERVAL '1' HOUR) AS window_end\n" +  // 1小时窗口结束时间
                "FROM dwd_trade_pay_detail_suc pay\n" +
                "LEFT JOIN order_detail_update upd\n" +
                "  ON pay.order_detail_id = upd.order_detail_id\n" +  // 按订单明细ID关联
                "  AND upd.update_time BETWEEN pay.callback_time AND pay.callback_time + INTERVAL '3' DAY\n" +  // 3天内的更新
                "  AND upd.new_source_type_code = '2402'\n" +  // 只关注更新为2402的记录
                "WHERE\n" +
                "    pay.source_type_code <> '2402'  -- 支付时source_type_code不是2402\n" +
                "    AND upd.order_id IS NULL  -- 3天内没有更新为2402的记录（即警告订单）\n" +
                "GROUP BY\n" +
                "    pay.shop_id,\n" +
                "    TUMBLE(pay.row_op_ts, INTERVAL '1' HOUR)");  // 按店铺和1小时窗口分组

        env.execute("DwsTradeWarningOrder");
    }
}
